/*
 * Copyright (c) 2016. 恒昌互联网运营中心.
 */
package com.credithc.eagleeye.queue;
import com.alibaba.fastjson.JSON;
import com.credithc.eagleeye.config.EagleMonitorRedisConfig;
import com.credithc.eagleeye.thread.ThreadFactoryImpl;
import com.credithc.eagleeye.vo.RpcTraceInfoVo;
import com.credithc.framework.redis.util.RedisUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
/**
 * 描 述: <br/>
 * 作 者: zhangjin<br/>
 * 创 建：2016年09月26日<br/>
 * 版 本：v1.0.0<br>
 * <p/>
 * 历 史: (版本) 作者 时间 注释 <br/>
 */
@Component
@Slf4j
public class EagleRedisQueueManager {
    @Autowired
    private EagleMonitorRedisConfig eagleMonitorRedisConfig;
    @Autowired
    private RedisUtil redisUtil;
    @Autowired
    private EagleTraceInfoQueueManager eagleTraceInfoQueueManager;
    private static AtomicBoolean initialized = new AtomicBoolean(false);


    private static ExecutorService singleThreadExecutor =new ThreadPoolExecutor(1,
            1,0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<Runnable>(),
            new ThreadFactoryImpl("fetchRpcTraceInfoFromRedis_")
    );
    private EagleRedisQueueManager() {
    }
    public synchronized void start()  {
        if( initialized.compareAndSet(false, true)) {
            if (eagleMonitorRedisConfig.isEnable()) {
                try {
                    singleThreadExecutor.submit(new EagleRedisQueueTask());
                    log.info(String.format("eagle-eye-transfer fetchRpcTraceInfo from redis thread is start!redis host:%s", eagleMonitorRedisConfig.getAddress()));
                } catch (Exception ex) {
                    log.error(String.format("eagle-eye-transfer fetchRpcTraceInfo from redis thread is error! exception:%s", ex.getMessage()));
                }
            }
        }
    }

    public synchronized void stop() {
        singleThreadExecutor.shutdownNow();
        log.info("eagle-eye-transfer fetchRpcTraceInfo from redis thread is stop!");
    }

    class EagleRedisQueueTask implements Runnable {
        private static final String EAGLE_EYE_REDIS_LIST_KEY = "eagle_eye_info_list";
        /**
         * 从redis获取数据后，如果放入本地内存队列失败，暂停秒数后重试
         */
        private final static int QUEUE_OFFER_EXCEPTION_SLEEP_SECONDS=3;

        @Override
        public void run() {
            boolean start = true;
            while (start) {
                try {
                    log.debug("开始获取redis中EagleEye数据.....");
                    String json = redisUtil.listPop(EAGLE_EYE_REDIS_LIST_KEY);
                    if(json!=null) {
                        RpcTraceInfoVo rpcTraceInfoVo = JSON.parseObject(json, RpcTraceInfoVo.class);
                        boolean success = eagleTraceInfoQueueManager.offer(rpcTraceInfoVo);
                        while (!success) {
                            TimeUnit.SECONDS.sleep(QUEUE_OFFER_EXCEPTION_SLEEP_SECONDS);
                            success = eagleTraceInfoQueueManager.offer(rpcTraceInfoVo);
                        }
                    }else{
                        Thread.sleep(500);
                    }
                } catch (Exception ex) {
                    log.error(String.format("获取redis中EagleEye数据 ERROR : %s", ex.getMessage()),ex);
                }
            }
        }
    }

}
